package com.gonsin.gmessage.logic.mqtt;

import com.gonsin.gmessage.annotations.GMessageEnvironment;
import com.gonsin.gmessage.logic.GMessage;
import com.gonsin.gmessage.logic.mqtt.config.MqttConfig;
import lombok.Data;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.mqttv5.client.*;
import org.eclipse.paho.mqttv5.client.persist.MemoryPersistence;
import org.eclipse.paho.mqttv5.common.MqttException;
import org.eclipse.paho.mqttv5.common.MqttMessage;
import org.eclipse.paho.mqttv5.common.MqttSubscription;
import org.eclipse.paho.mqttv5.common.packet.MqttProperties;

import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.HashSet;
import java.util.Set;

/**
 * 管理mqtt真实的连接对象，实现最基本的订阅、发布、连接、断开等功能
 */
@Slf4j
public class MqttClient implements MqttCallback {

    private MqttAsyncClient client;

    private MqttConfig config;

    private MemoryPersistence persistence = new MemoryPersistence();

    private Set<TopicHolder> holders = new HashSet<>();

    /**
     * 是否已连接
     */
    private boolean connected;

    public MqttClient(MqttConfig config) {
        this.config = config;
    }

    /**
     * 连接到mqtt
     */
    @SneakyThrows
    public void connect() {
        MqttConnectionOptions connOpts = new MqttConnectionOptions();

        // v5开始 clean session 改名叫  clean start
        connOpts.setCleanStart(false);
        connOpts.setUserName(config.getUsername());
        connOpts.setPassword(config.getPassword().getBytes(StandardCharsets.UTF_8));
        connOpts.setAutomaticReconnect(true);
        connOpts.setAutomaticReconnectDelay(1000, 3000);

        // v5 新增遗愿topic 延迟发送功能
//        MqttMessage willMessage = new MqttMessage();
//        willMessage.setQos(2);
//        willMessage.setPayload("一些内容".getBytes(StandardCharsets.UTF_8));
//        connOpts.setWill("/client/has/kill", willMessage);

        try {
            client = new MqttAsyncClient(config.getBroker(), config.getClientId(), persistence);
            client.setCallback(this);
            client.connect(connOpts).waitForCompletion();
            if(GMessageEnvironment.LOG){
                log.info("MQTT连接成功");
            }
        } catch (Exception ex) {
            log.error(ex.toString(), ex);
            Thread.sleep(1000);
            connect();
        }

        connected = true;
    }


    /**
     * 订阅 topic
     */
    public void subscribe(String topic, int qos, IMqttMessageListener mqttMessageListener)  {
        holders.add(new TopicHolder(topic, qos, mqttMessageListener));
        subscribeInner(topic, qos, mqttMessageListener);
    }

    @SneakyThrows
    private void subscribeInner(String topic, int qos, IMqttMessageListener mqttMessageListener){
        MqttSubscription subscription = new MqttSubscription(topic, qos);
        MqttProperties mqttProperties = new MqttProperties();
        mqttProperties.setSubscriptionIdentifiers(Collections.singletonList(0));

        if(GMessageEnvironment.LOG){
            log.info("MQTT 订阅 {}", topic);
        }
        try {
            client.subscribe(subscription, null, null, mqttMessageListener, mqttProperties).waitForCompletion();
        } catch (Exception ex) {
            log.error(ex.toString(), ex);
        }
    }


    /**
     * 发布 topic
     */
    @SneakyThrows
    public IMqttToken publish(String topic, int qos, String payload) {
        if(GMessageEnvironment.LOG){
            log.info("MQTT 发布 {}", topic);
        }
        MqttMessage message = new MqttMessage();
        message.setQos(qos);
        message.setRetained(false);
        if(qos > 0){
            MqttProperties properties = new MqttProperties();
            properties.setMessageExpiryInterval(60L);
            message.setProperties(properties);
        }
        if(payload != null){
            message.setPayload(payload.getBytes(StandardCharsets.UTF_8));
        }
        IMqttToken token = client.publish(topic, message);
        return token;
    }

    @SneakyThrows
    @Override
    public void disconnected(MqttDisconnectResponse disconnectResponse) {
//        connected = false;
//        Thread.sleep(1000);
//        if(client != null){
//            client.reconnect();
//        }
//        try {
//            connect();
//        } catch (Exception ex) {
//            log.error(ex.toString(), ex);
//        }
    }

    @Override
    public void mqttErrorOccurred(MqttException exception) {

    }

    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {

    }

    @Override
    public void deliveryComplete(IMqttToken token) {

    }

    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        if(reconnect){

            if(GMessageEnvironment.LOG){
                log.info("MQTT 重连成功，重新执行订阅 ");
            }
            // 重新订阅
            for(TopicHolder holder : holders){
                subscribeInner(holder.topic, holder.qos, holder.mqttMessageListener);
            }
        }
    }

    @Override
    public void authPacketArrived(int reasonCode, MqttProperties properties) {

    }

    @Data
    class TopicHolder {

        private String topic;

        private int qos;

        private IMqttMessageListener mqttMessageListener;

        public TopicHolder(String topic, int qos, IMqttMessageListener mqttMessageListener) {
            this.topic = topic;
            this.qos = qos;
            this.mqttMessageListener = mqttMessageListener;
        }
    }
}
